import { RedisClient, createClient } from "redis";
import { Observable, Subscriber } from "rxjs";

class RedisSubscriber {
  private host: string;
  private port: number;
  private client: RedisClient;

  constructor(host: string, port: number) {
    this.host = host;
    this.port = port;

    this.client = createClient(port, host);
  }

  subscribe<T>(channel: string): Observable<T> {
    const producer = (subscriber: Subscriber<T>) => {
      this.client.subscribe(channel);

      this.client.on("message", (channel: string, message: string) => {
        if (message) {
          subscriber.next(JSON.parse(message));
        }
      });

      this.client.on("unsubscribe", (channel: string, count: number) => {
        subscriber.complete();
      });
    };

    return new Observable<T>(producer);
  }

  async dispose() {
    const fn = (): void => {
      this.client.unsubscribe();
      this.client.quit();
    };

    await fn();
  }
}

export { RedisSubscriber };
